/*
 * Copyright (C) 2018 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.spanner.ddl;

import static com.google.common.truth.Truth.assertThat;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.text.IsEqualCompressingWhiteSpace.equalToCompressingWhiteSpace;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;

import com.google.cloud.spanner.Dialect;
import com.google.cloud.teleport.spanner.common.Type;
import com.google.cloud.teleport.spanner.ddl.ForeignKey.ReferentialAction;
import com.google.cloud.teleport.spanner.ddl.GraphElementTable.GraphNodeTableReference;
import com.google.cloud.teleport.spanner.ddl.GraphElementTable.LabelToPropertyDefinitions;
import com.google.cloud.teleport.spanner.ddl.GraphElementTable.PropertyDefinition;
import com.google.cloud.teleport.spanner.ddl.IndexColumn.IndexColumnsBuilder;
import com.google.cloud.teleport.spanner.ddl.IndexColumn.Order;
import com.google.cloud.teleport.spanner.ddl.Udf.SqlSecurity;
import com.google.cloud.teleport.spanner.proto.ExportProtos.Export;
import com.google.common.base.Optional;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import org.junit.Test;

/** Test coverage for {@link Ddl}. */
public class DdlTest {

  @Test
  public void emptyDb() {
    Ddl empty = Ddl.builder().build();
    assertThat(empty.allTables(), empty());
    assertThat(empty.prettyPrint(), equalTo(""));
    assertTrue(empty.equals(empty));
    assertNotNull(empty.hashCode());
  }

  @Test
  public void pgEmptyDb() {
    Ddl empty = Ddl.builder(Dialect.POSTGRESQL).build();
    assertEquals(empty.dialect(), Dialect.POSTGRESQL);
    assertThat(empty.allTables(), empty());
    assertThat(empty.prettyPrint(), equalTo(""));
    assertFalse(empty.equals(null));
    assertFalse(empty.equals(Boolean.TRUE));
    assertNotNull(empty.hashCode());
  }

  @Test
  public void simple() {
    Ddl.Builder builder = Ddl.builder();
    builder
        .createTable("Users")
        .column("id")
        .int64()
        .notNull()
        .endColumn()
        .column("gen_id")
        .int64()
        .notNull()
        .generatedAs("MOD(id+1, 64)")
        .stored()
        .endColumn()
        .column("first_name")
        .string()
        .size(10)
        .defaultExpression("'John'")
        .endColumn()
        .column("last_name")
        .type(Type.string())
        .max()
        .endColumn()
        .column("full_name")
        .type(Type.string())
        .max()
        .generatedAs("CONCAT(first_name, ' ', last_name)")
        .stored()
        .endColumn()
        .column("HiddenColumn")
        .type(Type.string())
        .max()
        .isHidden(true)
        .endColumn()
        .column("identity_column")
        .type(Type.int64())
        .isIdentityColumn(true)
        .sequenceKind("bit_reversed_positive")
        .counterStartValue(1000L)
        .skipRangeMin(2000L)
        .skipRangeMax(3000L)
        .endColumn()
        .column("identity_column_no_kind")
        .type(Type.int64())
        .isIdentityColumn(true)
        .counterStartValue(1000L)
        .skipRangeMin(2000L)
        .skipRangeMax(3000L)
        .endColumn()
        .primaryKey()
        .asc("id")
        .asc("gen_id")
        .end()
        .indexes(
            ImmutableList.of(
                "CREATE INDEX `UsersByFirstName` ON `Users` (`first_name`)",
                "CREATE INDEX `UsersByIdAndFirstName` ON `Users` (`id`,`gen_id`,`first_name`), INTERLEAVE IN `Users`"))
        .foreignKeys(
            ImmutableList.of(
                "ALTER TABLE `Users` ADD CONSTRAINT `fk` FOREIGN KEY (`first_name`)"
                    + " REFERENCES `AllowedNames` (`first_name`)",
                "ALTER TABLE `Users` ADD CONSTRAINT `fk_odc` FOREIGN KEY (`last_name`)"
                    + " REFERENCES `AllowedNames` (`last_name`) ON DELETE CASCADE",
                "ALTER TABLE `Users` ADD CONSTRAINT `fk_not_enforced_no_action` FOREIGN KEY (`last_name`) "
                    + "  REFERENCES `AllowedNames` (`last_name`) ON DELETE NO ACTION NOT ENFORCED",
                "ALTER TABLE `Users` ADD CONSTRAINT `fk_enforced` FOREIGN KEY (`last_name`) "
                    + "  REFERENCES `AllowedNames` (`last_name`) ENFORCED"))
        .checkConstraints(ImmutableList.of("CONSTRAINT `ck` CHECK (`first_name` != `last_name`)"))
        .endTable();
    Export export =
        Export.newBuilder()
            .addDatabaseOptions(
                Export.DatabaseOption.newBuilder()
                    .setOptionName("version_retention_period")
                    .setOptionValue("4d")
                    .setOptionType("STRING")
                    .build())
            .addDatabaseOptions(
                Export.DatabaseOption.newBuilder()
                    .setOptionName("default_sequence_kind")
                    .setOptionValue("bit_reversed_positive")
                    .setOptionType("STRING")
                    .build())
            .build();
    builder.mergeDatabaseOptions(export.getDatabaseOptionsList());
    Ddl ddl = builder.build();
    assertThat(
        ddl.prettyPrint(),
        equalToCompressingWhiteSpace(
            "ALTER DATABASE `%db_name%` SET OPTIONS ( version_retention_period = \"4d\" )"
                + " ALTER DATABASE `%db_name%` SET OPTIONS ( default_sequence_kind = \"bit_reversed_positive\" )"
                + " CREATE TABLE `Users` ("
                + " `id` INT64 NOT NULL,"
                + " `gen_id` INT64 NOT NULL AS (MOD(id+1, 64)) STORED,"
                + " `first_name` STRING(10) DEFAULT ('John'),"
                + " `last_name` STRING(MAX),"
                + " `full_name` STRING(MAX) AS (CONCAT(first_name, ' ', last_name)) STORED,"
                + " `HiddenColumn`                          STRING(MAX) HIDDEN,"
                + " `identity_column` INT64 GENERATED BY DEFAULT AS IDENTITY ("
                + "BIT_REVERSED_POSITIVE SKIP RANGE 2000, 3000 START COUNTER WITH 1000),"
                + " `identity_column_no_kind` INT64 GENERATED BY DEFAULT AS IDENTITY ("
                + "SKIP RANGE 2000, 3000 START COUNTER WITH 1000),"
                + " CONSTRAINT `ck` CHECK (`first_name` != `last_name`),"
                + " ) PRIMARY KEY (`id` ASC, `gen_id` ASC)"
                + " CREATE INDEX `UsersByFirstName` ON `Users` (`first_name`)"
                + " CREATE INDEX `UsersByIdAndFirstName` ON `Users` (`id`,`gen_id`,`first_name`), INTERLEAVE IN `Users`"
                + " ALTER TABLE `Users` ADD CONSTRAINT `fk` FOREIGN KEY (`first_name`)"
                + " REFERENCES `AllowedNames` (`first_name`)"
                + " ALTER TABLE `Users` ADD CONSTRAINT `fk_odc` FOREIGN KEY (`last_name`)"
                + " REFERENCES `AllowedNames` (`last_name`) ON DELETE CASCADE"
                + " ALTER TABLE `Users` ADD CONSTRAINT `fk_not_enforced_no_action`"
                + " FOREIGN KEY (`last_name`) REFERENCES "
                + "`AllowedNames` (`last_name`) ON DELETE NO ACTION NOT ENFORCED"
                + " ALTER TABLE `Users` ADD CONSTRAINT `fk_enforced`"
                + " FOREIGN KEY (`last_name`) REFERENCES "
                + "`AllowedNames` (`last_name`) ENFORCED"));
    List<String> statements = ddl.statements();
    assertEquals(9, statements.size());
    assertThat(
        statements.get(0),
        equalToCompressingWhiteSpace(
            " CREATE TABLE `Users` ("
                + " `id` INT64 NOT NULL,"
                + " `gen_id` INT64 NOT NULL AS (MOD(id+1, 64)) STORED,"
                + " `first_name` STRING(10) DEFAULT ('John'),"
                + " `last_name` STRING(MAX),"
                + " `full_name` STRING(MAX) AS (CONCAT(first_name, ' ', last_name)) STORED,"
                + " `HiddenColumn`                          STRING(MAX) HIDDEN,"
                + " `identity_column` INT64 GENERATED BY DEFAULT AS IDENTITY ("
                + "BIT_REVERSED_POSITIVE SKIP RANGE 2000, 3000 START COUNTER WITH 1000),"
                + " `identity_column_no_kind` INT64 GENERATED BY DEFAULT AS IDENTITY ("
                + "SKIP RANGE 2000, 3000 START COUNTER WITH 1000),"
                + " CONSTRAINT `ck` CHECK (`first_name` != `last_name`),"
                + " ) PRIMARY KEY (`id` ASC, `gen_id` ASC)"));
    assertThat(
        statements.get(1),
        equalToCompressingWhiteSpace(" CREATE INDEX `UsersByFirstName` ON `Users` (`first_name`)"));
    assertThat(
        statements.get(2),
        equalToCompressingWhiteSpace(
            " CREATE INDEX `UsersByIdAndFirstName` ON `Users` (`id`,`gen_id`,`first_name`), INTERLEAVE IN `Users`"));
    assertThat(
        statements.get(3),
        equalToCompressingWhiteSpace(
            "ALTER TABLE `Users` ADD CONSTRAINT `fk` FOREIGN KEY (`first_name`) REFERENCES"
                + " `AllowedNames` (`first_name`)"));
    assertThat(
        statements.get(4),
        equalToCompressingWhiteSpace(
            "ALTER TABLE `Users` ADD CONSTRAINT `fk_odc` FOREIGN KEY (`last_name`) REFERENCES"
                + " `AllowedNames` (`last_name`) ON DELETE CASCADE"));

    assertThat(
        statements.get(5),
        equalToCompressingWhiteSpace(
            "ALTER TABLE `Users` ADD CONSTRAINT `fk_not_enforced_no_action`"
                + " FOREIGN KEY (`last_name`) REFERENCES "
                + "`AllowedNames` (`last_name`) ON DELETE NO ACTION NOT ENFORCED"));
    assertThat(
        statements.get(6),
        equalToCompressingWhiteSpace(
            "ALTER TABLE `Users` ADD CONSTRAINT `fk_enforced`"
                + " FOREIGN KEY (`last_name`) REFERENCES "
                + "`AllowedNames` (`last_name`) ENFORCED"));
    assertThat(
        statements.get(7),
        equalToCompressingWhiteSpace(
            "ALTER DATABASE `%db_name%` SET OPTIONS ( version_retention_period = \"4d\" )"));
    assertThat(
        statements.get(8),
        equalToCompressingWhiteSpace(
            "ALTER DATABASE `%db_name%` SET OPTIONS ( default_sequence_kind = \"bit_reversed_positive\" )"));
    assertNotNull(ddl.hashCode());
  }

  @Test
  public void pgSimple() {
    Ddl.Builder builder = Ddl.builder(Dialect.POSTGRESQL);
    builder
        .createTable("Users")
        .column("id")
        .pgInt8()
        .notNull()
        .endColumn()
        .column("gen_id")
        .pgInt8()
        .notNull()
        .generatedAs("MOD(id+1, 64)")
        .stored()
        .endColumn()
        .column("gen_id_virtual")
        .pgInt8()
        .notNull()
        .generatedAs("MOD(id+1, 64)")
        .endColumn()
        .column("first_name")
        .pgVarchar()
        .size(10)
        .defaultExpression("John")
        .endColumn()
        .column("last_name")
        .type(Type.pgVarchar())
        .max()
        .defaultExpression("Lennon")
        .endColumn()
        .column("full_name")
        .type(Type.pgVarchar())
        .max()
        .generatedAs("CONCAT(first_name, ' ', last_name)")
        .stored()
        .endColumn()
        .column("identity_column")
        .pgInt8()
        .isIdentityColumn(true)
        .sequenceKind("bit_reversed_positive")
        .counterStartValue(1000L)
        .skipRangeMin(2000L)
        .skipRangeMax(3000L)
        .endColumn()
        .column("identity_column_no_kind")
        .pgInt8()
        .isIdentityColumn(true)
        .counterStartValue(1000L)
        .skipRangeMin(2000L)
        .skipRangeMax(3000L)
        .endColumn()
        .column("update_time")
        .pgSpannerCommitTimestamp()
        .notNull()
        .endColumn()
        .primaryKey()
        .asc("id")
        .asc("gen_id")
        .end()
        .indexes(
            ImmutableList.of("CREATE INDEX \"UsersByFirstName\" ON \"Users\" (\"first_name\")"))
        .foreignKeys(
            ImmutableList.of(
                "ALTER TABLE \"Users\" ADD CONSTRAINT \"fk\" FOREIGN KEY (\"first_name\")"
                    + " REFERENCES \"AllowedNames\" (\"first_name\")",
                "ALTER TABLE \"Users\" ADD CONSTRAINT \"fk_odc\" FOREIGN KEY (\"last_name\")"
                    + " REFERENCES \"AllowedNames\" (\"last_name\") ON DELETE CASCADE"))
        .checkConstraints(
            ImmutableList.of("CONSTRAINT \"ck\" CHECK (\"first_name\" != \"last_name\")"))
        .endTable();
    Export export =
        Export.newBuilder()
            .addDatabaseOptions(
                Export.DatabaseOption.newBuilder()
                    .setOptionName("version_retention_period")
                    .setOptionValue("4d")
                    .setOptionType("character varying")
                    .build())
            .addDatabaseOptions(
                Export.DatabaseOption.newBuilder()
                    .setOptionName("default_sequence_kind")
                    .setOptionValue("bit_reversed_positive")
                    .setOptionType("character varying")
                    .build())
            .build();
    builder.mergeDatabaseOptions(export.getDatabaseOptionsList());
    Ddl ddl = builder.build();
    assertThat(
        ddl.prettyPrint(),
        equalToCompressingWhiteSpace(
            "ALTER DATABASE \"%db_name%\" SET spanner.version_retention_period = '4d'"
                + " ALTER DATABASE \"%db_name%\" SET spanner.default_sequence_kind = 'bit_reversed_positive'"
                + " CREATE TABLE \"Users\" ("
                + " \"id\" bigint NOT NULL,"
                + " \"gen_id\" bigint NOT NULL GENERATED ALWAYS AS (MOD(id+1, 64)) STORED,"
                + " \"gen_id_virtual\" bigint NOT NULL GENERATED ALWAYS AS (MOD(id+1, 64)) VIRTUAL,"
                + " \"first_name\" character varying(10) DEFAULT John,"
                + " \"last_name\" character varying DEFAULT Lennon,"
                + " \"full_name\" character varying GENERATED ALWAYS AS"
                + " (CONCAT(first_name, ' ', last_name)) STORED,"
                + " \"identity_column\" bigint GENERATED BY DEFAULT AS IDENTITY ("
                + "BIT_REVERSED_POSITIVE SKIP RANGE 2000 3000 START COUNTER WITH 1000),"
                + " \"identity_column_no_kind\" bigint GENERATED BY DEFAULT AS IDENTITY ("
                + "SKIP RANGE 2000 3000 START COUNTER WITH 1000),"
                + " \"update_time\" spanner.commit_timestamp NOT NULL,"
                + " CONSTRAINT \"ck\" CHECK (\"first_name\" != \"last_name\"),"
                + " PRIMARY KEY (\"id\", \"gen_id\")"
                + " ) "
                + " CREATE INDEX \"UsersByFirstName\" ON \"Users\" (\"first_name\")"
                + " ALTER TABLE \"Users\" ADD CONSTRAINT \"fk\" FOREIGN KEY (\"first_name\")"
                + " REFERENCES \"AllowedNames\" (\"first_name\")"
                + " ALTER TABLE \"Users\" ADD CONSTRAINT \"fk_odc\" FOREIGN KEY (\"last_name\")"
                + " REFERENCES \"AllowedNames\" (\"last_name\") ON DELETE CASCADE"));
    assertNotNull(ddl.hashCode());
  }

  @Test
  public void embeddingVector() {
    Ddl.Builder builder = Ddl.builder();
    builder
        .createTable("Users")
        .column("id")
        .int64()
        .notNull()
        .endColumn()
        .column("embedding_vector")
        .type(Type.array(Type.float64()))
        .arrayLength(Integer.valueOf(128))
        .endColumn()
        .primaryKey()
        .asc("id")
        .end()
        .endTable();

    Ddl ddl = builder.build();
    assertThat(
        ddl.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE TABLE `Users` ("
                + " `id` INT64 NOT NULL,"
                + " `embedding_vector` ARRAY<FLOAT64>(vector_length=>128),"
                + " ) PRIMARY KEY (`id` ASC)"));
    assertNotEquals(ddl.hashCode(), 0);
  }

  @Test
  public void pgEmbeddingVector() {
    Ddl.Builder builder = Ddl.builder(Dialect.POSTGRESQL);
    builder
        .createTable("Users")
        .column("id")
        .pgInt8()
        .notNull()
        .endColumn()
        .column("embedding_vector")
        .type(Type.pgArray(Type.pgFloat8()))
        .arrayLength(Integer.valueOf(64))
        .endColumn()
        .primaryKey()
        .asc("id")
        .end()
        .endTable();

    Ddl ddl = builder.build();

    assertThat(
        ddl.prettyPrint(),
        equalToCompressingWhiteSpace(
            " CREATE TABLE \"Users\" ("
                + " \"id\" bigint NOT NULL,"
                + " \"embedding_vector\" double precision[] vector length 64,"
                + " PRIMARY KEY (\"id\")"
                + " ) "));
    assertNotEquals(ddl.hashCode(), 0);
  }

  @Test
  public void interleaves() {
    Ddl ddl =
        Ddl.builder()
            .createTable("Users")
            .column("id")
            .int64()
            .notNull()
            .endColumn()
            .column("first_name")
            .string()
            .size(10)
            .endColumn()
            .column("last_name")
            .type(Type.string())
            .max()
            .endColumn()
            .primaryKey()
            .asc("id")
            .end()
            .endTable()
            .createTable("Account")
            .column("id")
            .int64()
            .notNull()
            .endColumn()
            .column("balanceId")
            .int64()
            .notNull()
            .endColumn()
            .column("balance")
            .float64()
            .notNull()
            .endColumn()
            .primaryKey()
            .asc("id")
            .end()
            .interleaveInParent("Users")
            .onDeleteCascade()
            .endTable()
            .build();
    assertThat(
        ddl.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE TABLE `Users` ("
                + " `id`                                    INT64 NOT NULL,"
                + " `first_name`                            STRING(10),"
                + " `last_name`                             STRING(MAX),"
                + " ) PRIMARY KEY (`id` ASC)"
                + " CREATE TABLE `Account` ("
                + " `id`                                    INT64 NOT NULL,"
                + " `balanceId`                             INT64 NOT NULL,"
                + " `balance`                               FLOAT64 NOT NULL,"
                + " ) PRIMARY KEY (`id` ASC), "
                + " INTERLEAVE IN PARENT `Users` ON DELETE CASCADE"));
    Collection<Table> rootTables = ddl.rootTables();
    assertEquals(1, rootTables.size());
    assertEquals("Users", rootTables.iterator().next().name());
    HashMultimap<Integer, String> perLevelView = ddl.perLevelView();
    assertEquals(2, perLevelView.size());
    assertTrue(perLevelView.containsKey(0));
    assertEquals("users", perLevelView.get(0).iterator().next());
    assertTrue(perLevelView.containsKey(1));
    assertEquals("account", perLevelView.get(1).iterator().next());
    assertNotNull(ddl.hashCode());
  }

  @Test
  public void pgInterleaves() {
    Ddl ddl =
        Ddl.builder(Dialect.POSTGRESQL)
            .createTable("Users")
            .column("id")
            .pgInt8()
            .notNull()
            .endColumn()
            .column("first_name")
            .pgVarchar()
            .size(10)
            .endColumn()
            .column("last_name")
            .type(Type.pgVarchar())
            .max()
            .endColumn()
            .primaryKey()
            .asc("id")
            .end()
            .endTable()
            .createTable("Account")
            .column("id")
            .pgInt8()
            .notNull()
            .endColumn()
            .column("balanceId")
            .pgInt8()
            .notNull()
            .endColumn()
            .column("balance")
            .pgFloat8()
            .notNull()
            .endColumn()
            .primaryKey()
            .asc("id")
            .end()
            .interleaveInParent("Users")
            .onDeleteCascade()
            .endTable()
            .build();
    assertThat(
        ddl.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE TABLE \"Users\" ("
                + " \"id\"                                    bigint NOT NULL,"
                + " \"first_name\"                            character varying(10),"
                + " \"last_name\"                             character varying,"
                + " PRIMARY KEY (\"id\")"
                + " ) "
                + " CREATE TABLE \"Account\" ("
                + " \"id\"                                    bigint NOT NULL,"
                + " \"balanceId\"                             bigint NOT NULL,"
                + " \"balance\"                               double precision NOT NULL,"
                + " PRIMARY KEY (\"id\")"
                + " ) "
                + " INTERLEAVE IN PARENT \"Users\" ON DELETE CASCADE"));
    assertNotNull(ddl.hashCode());
  }

  @Test
  public void testDatabaseOptions() {
    Ddl.Builder builder = Ddl.builder();
    List<Export.DatabaseOption> dbOptionList = new ArrayList<>();
    dbOptionList.add(
        Export.DatabaseOption.newBuilder()
            .setOptionName("version_retention_period")
            .setOptionValue("4d")
            .setOptionType("STRING")
            .build());
    dbOptionList.add(
        Export.DatabaseOption.newBuilder()
            .setOptionName("optimizer_version")
            .setOptionValue("2")
            .build());
    builder.mergeDatabaseOptions(dbOptionList);
    Ddl ddl = builder.build();
    List<String> optionStatements = ddl.setOptionsStatements("database_id");
    assertThat(optionStatements.size(), is(1));
    assertThat(
        optionStatements.get(0),
        is("ALTER DATABASE `database_id` SET OPTIONS ( version_retention_period = \"4d\" )"));
    assertNotNull(ddl.hashCode());
  }

  @Test
  public void pgTestDatabaseOptions() {
    Ddl.Builder builder = Ddl.builder(Dialect.POSTGRESQL);
    List<Export.DatabaseOption> dbOptionList = new ArrayList<>();
    dbOptionList.add(
        Export.DatabaseOption.newBuilder()
            .setOptionName("version_retention_period")
            .setOptionValue("4d")
            .setOptionType("character varying")
            .build());
    dbOptionList.add(
        Export.DatabaseOption.newBuilder()
            .setOptionName("optimizer_version")
            .setOptionValue("2")
            .build());
    builder.mergeDatabaseOptions(dbOptionList);
    Ddl ddl = builder.build();
    List<String> optionStatements = ddl.setOptionsStatements("database_id");
    assertThat(optionStatements.size(), is(1));
    assertThat(
        optionStatements.get(0),
        is("ALTER DATABASE \"database_id\" SET spanner.version_retention_period = '4d'"));
    assertNotNull(ddl.hashCode());
  }

  @Test
  public void testIndex() {
    Index.Builder builder =
        Index.builder(Dialect.GOOGLE_STANDARD_SQL)
            .name("user_index")
            .table("User")
            .unique()
            .filter("`first_name` IS NOT NULL AND `last_name` IS NOT NULL");
    builder
        .columns()
        .create()
        .name("first_name")
        .asc()
        .endIndexColumn()
        .create()
        .name("last_name")
        .desc()
        .endIndexColumn()
        .create()
        .name("full_name")
        .storing()
        .endIndexColumn()
        .end();
    Index index = builder.build();
    assertThat(
        index.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE UNIQUE INDEX `user_index` ON `User`(`first_name` ASC,"
                + " `last_name` DESC) STORING (`full_name`) WHERE `first_name` IS"
                + " NOT NULL AND `last_name` IS NOT NULL"));
    assertTrue(index.equals(index));
    assertFalse(index.equals(Boolean.TRUE));
    builder = index.autoToBuilder();
    builder
        .columns()
        .create()
        .name("first_name")
        .asc()
        .endIndexColumn()
        .create()
        .name("last_name")
        .desc()
        .endIndexColumn()
        .create()
        .name("full_name")
        .storing()
        .endIndexColumn()
        .end();
    Index index1 = builder.build();
    assertTrue(index.equals(index1));
    assertNotNull(index.hashCode());
  }

  @Test
  public void testSearchIndex() {
    Index.Builder builder =
        Index.builder(Dialect.GOOGLE_STANDARD_SQL)
            .name("SearchIndex")
            .type("SEARCH")
            .table("Messages")
            .interleaveIn("Users")
            .partitionBy(ImmutableList.of("UserId"))
            .options(ImmutableList.of("sort_order_sharding=TRUE"));
    builder
        .columns()
        .create()
        .name("Subject_Tokens")
        .none()
        .endIndexColumn()
        .create()
        .name("Body_Tokens")
        .none()
        .endIndexColumn()
        .create()
        .name("Data")
        .storing()
        .endIndexColumn()
        .end();
    Index index = builder.build();
    assertThat(
        index.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE SEARCH INDEX `SearchIndex` ON `Messages`(`Subject_Tokens` , `Body_Tokens` )"
                + " STORING (`Data`) PARTITION BY `UserId`, INTERLEAVE IN `Users` OPTIONS (sort_order_sharding=TRUE)"));
  }

  @Test
  public void testpgSearchIndex() {
    Index.Builder builder =
        Index.builder(Dialect.POSTGRESQL)
            .name("SearchIndex")
            .type("SEARCH")
            .table("Messages")
            .interleaveIn("Users")
            .partitionBy(ImmutableList.of("userid"))
            .orderBy(ImmutableList.of("orderid"))
            .options(ImmutableList.of("sort_order_sharding=TRUE"));
    builder
        .columns()
        .create()
        .name("subject_tokens")
        .none()
        .endIndexColumn()
        .create()
        .name("body_tokens")
        .none()
        .endIndexColumn()
        .create()
        .name("data")
        .storing()
        .endIndexColumn()
        .end();
    Index index = builder.build();
    assertThat(
        index.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE SEARCH INDEX \"SearchIndex\" ON \"Messages\"(\"subject_tokens\" , \"body_tokens\" )"
                + " INCLUDE (\"data\") PARTITION BY \"userid\" ORDER BY \"orderid\" INTERLEAVE IN \"Users\" WITH (sort_order_sharding=TRUE)"));
  }

  @Test
  public void testVectorIndex() {
    Index.Builder builder =
        Index.builder(Dialect.GOOGLE_STANDARD_SQL)
            .name("VectorIndex")
            .type("VECTOR")
            .table("Base")
            .options(ImmutableList.of("distance_type=\"COSINE\""));
    builder.columns().create().name("Embeddings").none().endIndexColumn().end();
    Index index = builder.build();
    assertThat(
        index.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE VECTOR INDEX `VectorIndex` ON `Base`(`Embeddings` ) OPTIONS (distance_type=\"COSINE\")"));
  }

  @Test
  public void testpgVectorIndex() {
    Index.Builder builder =
        Index.builder(Dialect.POSTGRESQL)
            .name("VectorIndex")
            .type("ScaNN")
            .table("Base")
            .options(ImmutableList.of("distance_type='COSINE'"));
    builder.columns().create().name("Embeddings").none().endIndexColumn().end();
    Index index = builder.build();
    assertThat(
        index.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE INDEX \"VectorIndex\" ON \"Base\" USING ScaNN (\"Embeddings\" ) WITH (distance_type='COSINE')"));
  }

  @Test
  public void pgTestIndex() {
    Index.Builder builder =
        Index.builder(Dialect.POSTGRESQL)
            .name("user_index")
            .table("User")
            .unique()
            .filter("\"first_name\" IS NOT NULL AND \"last_name\" IS NOT NULL");
    builder
        .columns()
        .create()
        .name("first_name")
        .asc()
        .endIndexColumn()
        .create()
        .name("last_name")
        .desc()
        .endIndexColumn()
        .create()
        .name("full_name")
        .storing()
        .endIndexColumn()
        .end();
    Index index = builder.build();
    assertThat(
        index.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE UNIQUE INDEX \"user_index\" ON \"User\"(\"first_name\" ASC,"
                + " \"last_name\" DESC) INCLUDE (\"full_name\") WHERE \"first_name\" IS"
                + " NOT NULL AND \"last_name\" IS NOT NULL"));
    assertTrue(index.equals(index));
    assertFalse(index.equals(Boolean.TRUE));
    builder = index.autoToBuilder();
    builder
        .columns()
        .create()
        .name("first_name")
        .asc()
        .endIndexColumn()
        .create()
        .name("last_name")
        .desc()
        .endIndexColumn()
        .create()
        .name("full_name")
        .storing()
        .endIndexColumn()
        .end();
    Index index1 = builder.build();
    assertTrue(index.equals(index1));
    assertNotNull(index.hashCode());
  }

  @Test
  public void pgTestIndexNullsOrder() {
    Index.Builder builder =
        Index.builder(Dialect.POSTGRESQL).name("user_index").table("User").unique();
    builder
        .columns()
        .create()
        .name("first_name")
        .asc()
        .nullsFirst()
        .endIndexColumn()
        .create()
        .name("last_name")
        .desc()
        .nullsLast()
        .endIndexColumn()
        .create()
        .name("first_nick_name")
        .asc()
        .nullsLast()
        .endIndexColumn()
        .create()
        .name("last_nick_name")
        .desc()
        .nullsFirst()
        .endIndexColumn()
        .create()
        .name("full_name")
        .storing()
        .endIndexColumn()
        .end();
    Index index = builder.build();
    assertThat(
        index.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE UNIQUE INDEX \"user_index\" ON \"User\"(\"first_name\" ASC NULLS FIRST,"
                + " \"last_name\" DESC NULLS LAST, \"first_nick_name\" ASC NULLS LAST,"
                + " \"last_nick_name\" DESC NULLS FIRST) INCLUDE (\"full_name\")"));
    assertNotNull(index.hashCode());
  }

  @Test
  public void pgTestCheckConstraint() {
    CheckConstraint checkConstraint =
        CheckConstraint.builder(Dialect.POSTGRESQL)
            .name("name_check")
            .expression("\"first_name\" != \"last_name\"")
            .build();
    assertThat(
        checkConstraint.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CONSTRAINT \"name_check\" CHECK (\"first_name\" != \"last_name\")"));
    assertNotNull(checkConstraint.hashCode());
  }

  @Test
  public void pgTestForeignKey() {
    ForeignKey.Builder builder =
        ForeignKey.builder(Dialect.POSTGRESQL)
            .name("account_to_user")
            .table("Account")
            .referencedTable("User");
    builder.columnsBuilder().add("account_id", "owner_name");
    builder.referencedColumnsBuilder().add("user_id", "full_name");
    ForeignKey foreignKey = builder.build();
    assertThat(
        foreignKey.prettyPrint(),
        equalToCompressingWhiteSpace(
            "ALTER TABLE \"Account\" ADD CONSTRAINT \"account_to_user\" FOREIGN KEY"
                + " (\"account_id\", \"owner_name\") REFERENCES \"User\" (\"user_id\","
                + " \"full_name\")"));
    // TODO(gunjj) Consider moving FK ON DELETE CASCADE test from simple()/pgSimple() to this method
    // for PG and for a separate such method for GSQL
    assertNotNull(foreignKey.hashCode());
  }

  @Test
  public void testModel() {
    Model.Builder model =
        Model.builder()
            .name("user_model")
            .remote(true)
            .options(ImmutableList.of("endpoint=\"test\""));

    model
        .inputColumn("i1")
        .type(Type.bool())
        .size(-1)
        .columnOptions(ImmutableList.of("required=FALSE"))
        .endInputColumn();
    model.inputColumn("i2").type(Type.string()).size(-1).endInputColumn();
    model
        .outputColumn("o1")
        .type(Type.struct(Type.StructField.of("a", Type.int64())))
        .size(-1)
        .columnOptions(ImmutableList.of("required=TRUE"))
        .endOutputColumn();
    model.outputColumn("o2").type(Type.float64()).size(-1).endOutputColumn();

    assertThat(
        model.build().prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE MODEL `user_model`"
                + " INPUT ( `i1` BOOL OPTIONS (required=FALSE), `i2` STRING(MAX), )"
                + " OUTPUT ( `o1` STRUCT<a INT64> OPTIONS (required=TRUE), `o2` FLOAT64, )"
                + " REMOTE OPTIONS (endpoint=\"test\")"));
  }

  @Test
  public void pgTestModel() {
    Model.Builder model =
        Model.builder(Dialect.POSTGRESQL)
            .name("user_model")
            .remote(true)
            .options(ImmutableList.of("endpoint=\"test\""));

    model
        .inputColumn("i1")
        .type(Type.bool())
        .size(-1)
        .columnOptions(ImmutableList.of("required=FALSE"))
        .endInputColumn();
    model.inputColumn("i2").type(Type.string()).size(-1).endInputColumn();
    model
        .outputColumn("o1")
        .type(Type.int64())
        .size(-1)
        .columnOptions(ImmutableList.of("required=TRUE"))
        .endOutputColumn();
    model.outputColumn("o2").type(Type.float64()).size(-1).endOutputColumn();

    assertThrows(IllegalArgumentException.class, () -> model.build().prettyPrint());
  }

  @Test
  public void testPropertyGraph() {
    // Craft Property Declarations
    PropertyGraph.PropertyDeclaration propertyDeclaration1 =
        new PropertyGraph.PropertyDeclaration("dummy-prop-name", "dummy-prop-type");
    PropertyGraph.PropertyDeclaration propertyDeclaration2 =
        new PropertyGraph.PropertyDeclaration("aliased-prop-name", "dummy-prop-type");
    ImmutableList<String> propertyDeclsLabel1 =
        ImmutableList.copyOf(Arrays.asList(propertyDeclaration1.name, propertyDeclaration2.name));

    // Craft Labels and associated property definitions
    PropertyGraph.GraphElementLabel label1 =
        new PropertyGraph.GraphElementLabel("dummy-label-name1", propertyDeclsLabel1);
    GraphElementTable.PropertyDefinition propertyDefinition1 =
        new PropertyDefinition("dummy-prop-name", "dummy-prop-name");
    GraphElementTable.PropertyDefinition propertyDefinition2 =
        new PropertyDefinition(
            "aliased-prop-name", "CONCAT(CAST(test_col AS STRING), \":\", \"dummy-column\")");
    GraphElementTable.LabelToPropertyDefinitions labelToPropertyDefinitions1 =
        new LabelToPropertyDefinitions(
            label1.name, ImmutableList.of(propertyDefinition1, propertyDefinition2));

    PropertyGraph.GraphElementLabel label2 =
        new PropertyGraph.GraphElementLabel("dummy-label-name2", ImmutableList.of());
    GraphElementTable.LabelToPropertyDefinitions labelToPropertyDefinitions2 =
        new LabelToPropertyDefinitions(label2.name, ImmutableList.of());

    PropertyGraph.GraphElementLabel label3 =
        new PropertyGraph.GraphElementLabel("dummy-label-name3", ImmutableList.of());
    GraphElementTable.LabelToPropertyDefinitions labelToPropertyDefinitions3 =
        new LabelToPropertyDefinitions(label3.name, ImmutableList.of());

    // Craft Node table
    GraphElementTable.Builder testNodeTable =
        GraphElementTable.builder()
            .baseTableName("base-table")
            .name("node-alias")
            .kind(GraphElementTable.Kind.NODE)
            .keyColumns(ImmutableList.of("primary-key"))
            .labelToPropertyDefinitions(
                ImmutableList.of(labelToPropertyDefinitions1, labelToPropertyDefinitions2));

    // Craft Edge table
    GraphElementTable.Builder testEdgeTable =
        GraphElementTable.builder()
            .baseTableName("edge-base-table")
            .name("edge-alias")
            .kind(GraphElementTable.Kind.EDGE)
            .keyColumns(ImmutableList.of("edge-primary-key"))
            .sourceNodeTable(
                new GraphNodeTableReference(
                    "base-table",
                    ImmutableList.of("node-key"),
                    ImmutableList.of("source-edge-key")))
            .targetNodeTable(
                new GraphNodeTableReference(
                    "base-table",
                    ImmutableList.of("other-node-key"),
                    ImmutableList.of("dest-edge-key")))
            .labelToPropertyDefinitions(ImmutableList.of(labelToPropertyDefinitions3));

    // Build PropertyGraph
    PropertyGraph.Builder propertyGraph =
        PropertyGraph.builder()
            .name("test-graph")
            .addLabel(label1)
            .addLabel(label2)
            .addLabel(label3)
            .addPropertyDeclaration(propertyDeclaration1)
            .addPropertyDeclaration(propertyDeclaration2)
            .addNodeTable(testNodeTable.autoBuild())
            .addEdgeTable(testEdgeTable.autoBuild());

    assertThat(
        propertyGraph.build().prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE PROPERTY GRAPH test-graph "
                + "NODE TABLES(\n"
                + "base-table AS node-alias\n"
                + " KEY (primary-key)\n"
                + "LABEL dummy-label-name1 "
                + "PROPERTIES(dummy-prop-name, CONCAT(CAST(test_col AS STRING), \":\", \"dummy-column\") AS aliased-prop-name)\n"
                + "LABEL dummy-label-name2 NO PROPERTIES)\n"
                + "EDGE TABLES(\n"
                + "edge-base-table AS edge-alias\n"
                + " KEY (edge-primary-key)\n"
                + "SOURCE KEY(source-edge-key) REFERENCES base-table(node-key) DESTINATION KEY(dest-edge-key) REFERENCES base-table(other-node-key)\n"
                + "LABEL dummy-label-name3 NO PROPERTIES"
                + ")"));
  }

  @Test
  public void testDynamicPropertyGraph() {
    // Craft Property Declarations
    PropertyGraph.PropertyDeclaration propertyDeclaration1 =
        new PropertyGraph.PropertyDeclaration("dummy-prop-name", "dummy-prop-type");
    PropertyGraph.PropertyDeclaration propertyDeclaration2 =
        new PropertyGraph.PropertyDeclaration("aliased-prop-name", "dummy-prop-type");
    ImmutableList<String> propertyDeclsLabel1 =
        ImmutableList.copyOf(Arrays.asList(propertyDeclaration1.name, propertyDeclaration2.name));

    // Craft Labels and associated property definitions
    PropertyGraph.GraphElementLabel label1 =
        new PropertyGraph.GraphElementLabel("dummy-label-name1", propertyDeclsLabel1);
    GraphElementTable.PropertyDefinition propertyDefinition1 =
        new PropertyDefinition("dummy-prop-name", "dummy-prop-name");
    GraphElementTable.PropertyDefinition propertyDefinition2 =
        new PropertyDefinition(
            "aliased-prop-name", "CONCAT(CAST(test_col AS STRING), \":\", \"dummy-column\")");
    GraphElementTable.LabelToPropertyDefinitions labelToPropertyDefinitions1 =
        new LabelToPropertyDefinitions(
            label1.name, ImmutableList.of(propertyDefinition1, propertyDefinition2));

    PropertyGraph.GraphElementLabel label2 =
        new PropertyGraph.GraphElementLabel("dummy-label-name2", ImmutableList.of());
    GraphElementTable.LabelToPropertyDefinitions labelToPropertyDefinitions2 =
        new LabelToPropertyDefinitions(label2.name, ImmutableList.of());

    PropertyGraph.GraphElementLabel label3 =
        new PropertyGraph.GraphElementLabel("dummy-label-name3", ImmutableList.of());
    GraphElementTable.LabelToPropertyDefinitions labelToPropertyDefinitions3 =
        new LabelToPropertyDefinitions(label3.name, ImmutableList.of());

    // Craft Node table
    GraphElementTable.Builder testNodeTable =
        GraphElementTable.builder()
            .baseTableName("base-table")
            .name("node-alias")
            .kind(GraphElementTable.Kind.NODE)
            .keyColumns(ImmutableList.of("primary-key"))
            .dynamicLabelExpression(
                new PropertyGraph.GraphDynamicLabelExpression("dynamic-label-column"))
            .labelToPropertyDefinitions(
                ImmutableList.of(labelToPropertyDefinitions1, labelToPropertyDefinitions2));

    // Craft Edge table
    GraphElementTable.Builder testEdgeTable =
        GraphElementTable.builder()
            .baseTableName("edge-base-table")
            .name("edge-alias")
            .kind(GraphElementTable.Kind.EDGE)
            .keyColumns(ImmutableList.of("edge-primary-key"))
            .dynamicPropertiesExpression(
                new PropertyGraph.GraphDynamicPropertiesExpression("dynamic-properties-column"))
            .sourceNodeTable(
                new GraphNodeTableReference(
                    "base-table",
                    ImmutableList.of("node-key"),
                    ImmutableList.of("source-edge-key")))
            .targetNodeTable(
                new GraphNodeTableReference(
                    "base-table",
                    ImmutableList.of("other-node-key"),
                    ImmutableList.of("dest-edge-key")))
            .labelToPropertyDefinitions(ImmutableList.of(labelToPropertyDefinitions3));

    // Build PropertyGraph
    PropertyGraph.Builder propertyGraph =
        PropertyGraph.builder()
            .name("test-graph")
            .addLabel(label1)
            .addLabel(label2)
            .addLabel(label3)
            .addPropertyDeclaration(propertyDeclaration1)
            .addPropertyDeclaration(propertyDeclaration2)
            .addNodeTable(testNodeTable.autoBuild())
            .addEdgeTable(testEdgeTable.autoBuild());

    assertThat(
        propertyGraph.build().prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE PROPERTY GRAPH test-graph "
                + "NODE TABLES(\n"
                + "base-table AS node-alias\n"
                + " KEY (primary-key)\n"
                + "LABEL dummy-label-name1 "
                + "PROPERTIES(dummy-prop-name, CONCAT(CAST(test_col AS STRING), \":\", \"dummy-column\") AS aliased-prop-name)\n"
                + "LABEL dummy-label-name2 NO PROPERTIES\n"
                + "DYNAMIC LABEL(dynamic-label-column))\n"
                + "EDGE TABLES(\n"
                + "edge-base-table AS edge-alias\n"
                + " KEY (edge-primary-key)\n"
                + "SOURCE KEY(source-edge-key) REFERENCES base-table(node-key) DESTINATION KEY(dest-edge-key) REFERENCES base-table(other-node-key)\n"
                + "LABEL dummy-label-name3 NO PROPERTIES\n"
                + "DYNAMIC PROPERTIES(dynamic-properties-column)"
                + ")"));
  }

  @Test
  public void testView() {
    View view = View.builder().name("user_view").query("SELECT * FROM `User`").build();
    assertThat(
        view.prettyPrint(),
        equalToCompressingWhiteSpace("CREATE VIEW `user_view` AS SELECT * FROM `User`"));
  }

  @Test
  public void pgTestView() {
    View view =
        View.builder(Dialect.POSTGRESQL)
            .name("user_view")
            .query("SELECT * FROM \"User\"")
            .security(View.SqlSecurity.INVOKER)
            .build();
    assertThat(
        view.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE VIEW \"user_view\" SQL SECURITY INVOKER AS SELECT * FROM \"User\""));
    Ddl.Builder ddlBuilder = Ddl.builder();
    ddlBuilder.addView(view);
    Ddl ddl = ddlBuilder.build();
    assertEquals(view, ddl.view("user_view"));
    List<String> statements = ddl.statements();
    assertEquals(1, statements.size());
    assertEquals(
        "CREATE VIEW \"user_view\" SQL SECURITY INVOKER AS SELECT * FROM \"User\"",
        statements.get(0));
    assertNotNull(view.hashCode());
  }

  @Test
  public void changeStreams() {
    Ddl ddl =
        Ddl.builder()
            .createChangeStream("ChangeStreamAll")
            .forClause("FOR ALL")
            .options(
                ImmutableList.of(
                    "retention_period=\"7d\"", "value_capture_type=\"OLD_AND_NEW_VALUES\""))
            .endChangeStream()
            .createChangeStream("ChangeStreamEmpty")
            .endChangeStream()
            .createChangeStream("ChangeStreamTableColumns")
            .forClause("FOR `T1`, `T2`(`c1`, `c2`), `T3`()")
            .endChangeStream()
            .build();
    assertThat(
        ddl.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE CHANGE STREAM `ChangeStreamAll`"
                + " FOR ALL"
                + " OPTIONS (retention_period=\"7d\", value_capture_type=\"OLD_AND_NEW_VALUES\")"
                + " CREATE CHANGE STREAM `ChangeStreamEmpty`"
                + " CREATE CHANGE STREAM `ChangeStreamTableColumns`"
                + " FOR `T1`, `T2`(`c1`, `c2`), `T3`()"));

    List<String> statements = ddl.statements();
    assertEquals(3, statements.size());
    assertThat(
        statements.get(0),
        equalToCompressingWhiteSpace(
            "CREATE CHANGE STREAM `ChangeStreamAll`"
                + " FOR ALL"
                + " OPTIONS (retention_period=\"7d\", value_capture_type=\"OLD_AND_NEW_VALUES\")"));
    assertThat(
        statements.get(1),
        equalToCompressingWhiteSpace(" CREATE CHANGE STREAM `ChangeStreamEmpty`"));
    assertThat(
        statements.get(2),
        equalToCompressingWhiteSpace(
            " CREATE CHANGE STREAM `ChangeStreamTableColumns`"
                + " FOR `T1`, `T2`(`c1`, `c2`), `T3`()"));
    assertNotNull(ddl.hashCode());
  }

  @Test
  public void pgChangeStreams() {
    Ddl ddl =
        Ddl.builder(Dialect.POSTGRESQL)
            .createChangeStream("ChangeStreamAll")
            .forClause("FOR ALL")
            .options(
                ImmutableList.of(
                    "retention_period='7d'", "value_capture_type='OLD_AND_NEW_VALUES'"))
            .endChangeStream()
            .createChangeStream("ChangeStreamEmpty")
            .endChangeStream()
            .createChangeStream("ChangeStreamTableColumns")
            .forClause("FOR \"T1\", \"T2\"(\"c1\", \"c2\"), \"T3\"()")
            .endChangeStream()
            .build();
    assertThat(
        ddl.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE CHANGE STREAM \"ChangeStreamAll\""
                + " FOR ALL"
                + " WITH (retention_period='7d', value_capture_type='OLD_AND_NEW_VALUES')"
                + " CREATE CHANGE STREAM \"ChangeStreamEmpty\""
                + " CREATE CHANGE STREAM \"ChangeStreamTableColumns\""
                + " FOR \"T1\", \"T2\"(\"c1\", \"c2\"), \"T3\"()"));

    List<String> statements = ddl.statements();
    assertEquals(3, statements.size());
    assertThat(
        statements.get(0),
        equalToCompressingWhiteSpace(
            "CREATE CHANGE STREAM \"ChangeStreamAll\""
                + " FOR ALL"
                + " WITH (retention_period='7d', value_capture_type='OLD_AND_NEW_VALUES')"));
    assertThat(
        statements.get(1),
        equalToCompressingWhiteSpace(" CREATE CHANGE STREAM \"ChangeStreamEmpty\""));
    assertThat(
        statements.get(2),
        equalToCompressingWhiteSpace(
            " CREATE CHANGE STREAM \"ChangeStreamTableColumns\""
                + " FOR \"T1\", \"T2\"(\"c1\", \"c2\"), \"T3\"()"));
    assertNotNull(ddl.hashCode());
  }

  @Test
  public void udfs() {
    Ddl.Builder ddlBuilder =
        Ddl.builder()
            .createUdf("spanner.Foo1")
            .dialect(Dialect.GOOGLE_STANDARD_SQL)
            .name("Foo1")
            .definition("(SELECT 'bar')")
            .endUdf()
            .createUdf("spanner.Foo2")
            .dialect(Dialect.GOOGLE_STANDARD_SQL)
            .name("Foo2")
            .definition("(SELECT 'bar')")
            .security(SqlSecurity.INVOKER)
            .type("STRING")
            .addParameter(
                UdfParameter.parse("arg0 STRING", "spanner.Foo", Dialect.GOOGLE_STANDARD_SQL))
            .addParameter(
                UdfParameter.parse(
                    "arg1 STRING DEFAULT 'bar'", "spanner.Foo", Dialect.GOOGLE_STANDARD_SQL))
            .endUdf();
    assertThat(ddlBuilder.hasUdf("spanner.Foo1"));
    assertThat(ddlBuilder.createUdf("spanner.Foo1").name().equals("Foo1"));
    Ddl ddl = ddlBuilder.build();

    String expectedDdlString =
        "\nCREATE FUNCTION `Foo1`() AS ((SELECT 'bar'))\n"
            + "CREATE FUNCTION `Foo2`(`arg0` STRING, `arg1` STRING DEFAULT 'bar')"
            + " RETURNS STRING SQL SECURITY INVOKER AS ((SELECT 'bar'))";
    assertThat(ddl.prettyPrint(), equalToCompressingWhiteSpace(expectedDdlString));

    List<String> statements = ddl.statements();
    assertEquals(2, statements.size());
    assertThat(
        statements.get(0),
        equalToCompressingWhiteSpace("CREATE FUNCTION `Foo1`() AS ((SELECT 'bar'))"));
    assertThat(
        statements.get(1),
        equalToCompressingWhiteSpace(
            "CREATE FUNCTION `Foo2`(`arg0` STRING, `arg1` STRING DEFAULT 'bar')"
                + " RETURNS STRING SQL SECURITY INVOKER AS ((SELECT 'bar'))"));
    assertNotNull(ddl.hashCode());

    assertThat(
        ddl.toBuilder().build().prettyPrint(), equalToCompressingWhiteSpace(expectedDdlString));
  }

  @Test
  public void sequences() {
    Ddl ddl =
        Ddl.builder()
            .createSequence("MySequence")
            .options(
                ImmutableList.of(
                    "sequence_kind=\"bit_reversed_positive\"",
                    "skip_range_min=0",
                    "skip_range_max=1000",
                    "start_with_counter=50"))
            .endSequence()
            .createSequence("MySequence2")
            .options(
                ImmutableList.of(
                    "skip_range_min=0", "skip_range_max=1000", "start_with_counter=50"))
            .endSequence()
            .build();
    assertThat(
        ddl.prettyPrint(),
        equalToCompressingWhiteSpace(
            "CREATE SEQUENCE `MySequence`"
                + " OPTIONS (sequence_kind=\"bit_reversed_positive\", "
                + " skip_range_min=0, skip_range_max=1000, start_with_counter=50)"
                + " CREATE SEQUENCE `MySequence2`"
                + " OPTIONS (skip_range_min=0, skip_range_max=1000, start_with_counter=50)"));

    List<String> statements = ddl.statements();
    assertEquals(2, statements.size());
    assertThat(
        statements.get(0),
        equalToCompressingWhiteSpace(
            "CREATE SEQUENCE `MySequence`"
                + " OPTIONS (sequence_kind=\"bit_reversed_positive\", "
                + " skip_range_min=0, skip_range_max=1000, start_with_counter=50)"));
    assertThat(
        statements.get(1),
        equalToCompressingWhiteSpace(
            "CREATE SEQUENCE `MySequence2`"
                + " OPTIONS (skip_range_min=0, skip_range_max=1000, start_with_counter=50)"));
    assertNotNull(ddl.hashCode());
  }

  @Test
  public void pgSequences() {
    Ddl ddl =
        Ddl.builder(Dialect.POSTGRESQL)
            .createSequence("MyPGSequence")
            .sequenceKind("bit_reversed_positive")
            .counterStartValue(Long.valueOf(30))
            .skipRangeMin(Long.valueOf(1))
            .skipRangeMax(Long.valueOf(1000))
            .endSequence()
            .createSequence("MyPGSequence2")
            .sequenceKind("bit_reversed_positive")
            .endSequence()
            .createSequence("MyPGSequence3")
            .counterStartValue(Long.valueOf(30))
            .skipRangeMin(Long.valueOf(1))
            .skipRangeMax(Long.valueOf(1000))
            .endSequence()
            .build();
    assertThat(
        ddl.prettyPrint(),
        equalToCompressingWhiteSpace(
            "\nCREATE SEQUENCE \"MyPGSequence\" BIT_REVERSED_POSITIVE"
                + " SKIP RANGE 1 1000 START COUNTER WITH 30 "
                + "\nCREATE SEQUENCE \"MyPGSequence2\" BIT_REVERSED_POSITIVE"
                + "\nCREATE SEQUENCE \"MyPGSequence3\" SKIP RANGE 1 1000 START COUNTER WITH 30"));

    List<String> statements = ddl.statements();
    assertEquals(3, statements.size());
    assertThat(
        statements.get(0),
        equalToCompressingWhiteSpace(
            "CREATE SEQUENCE \"MyPGSequence\" BIT_REVERSED_POSITIVE"
                + " SKIP RANGE 1 1000 START COUNTER WITH 30"));
    assertThat(
        statements.get(1),
        equalToCompressingWhiteSpace("CREATE SEQUENCE \"MyPGSequence2\" BIT_REVERSED_POSITIVE"));
    assertThat(
        statements.get(2),
        equalToCompressingWhiteSpace(
            "CREATE SEQUENCE \"MyPGSequence3\" SKIP RANGE 1 1000 START COUNTER WITH 30"));
    assertNotNull(ddl.hashCode());
  }

  @Test
  public void protobundle() {
    Ddl.Builder builder = Ddl.builder();
    builder.mergeProtoBundle(
        ImmutableSet.of(
            "com.google.cloud.teleport.spanner.tests.TestMessage",
            "com.google.cloud.teleport.spanner.tests.Order",
            "com.google.cloud.teleport.spanner.tests.Order.Item",
            "com.google.cloud.teleport.spanner.tests.Order.Address",
            "com.google.cloud.teleport.spanner.tests.Order.PaymentMode",
            "com.google.cloud.teleport.spanner.tests.OrderHistory",
            "com.google.cloud.teleport.spanner.tests.TestEnum"));
    Ddl ddl = builder.build();
    String expectedProtoBundle =
        "CREATE PROTO BUNDLE ("
            + "\n\t`com.google.cloud.teleport.spanner.tests.TestMessage`,"
            + "\n\t`com.google.cloud.teleport.spanner.tests.Order.PaymentMode`,"
            + "\n\t`com.google.cloud.teleport.spanner.tests.Order.Item`,"
            + "\n\t`com.google.cloud.teleport.spanner.tests.Order.Address`,"
            + "\n\t`com.google.cloud.teleport.spanner.tests.Order`,"
            + "\n\t`com.google.cloud.teleport.spanner.tests.TestEnum`,"
            + "\n\t`com.google.cloud.teleport.spanner.tests.OrderHistory`,)";
    assertThat(ddl.prettyPrint(), equalToCompressingWhiteSpace(expectedProtoBundle));

    List<String> statements = ddl.statements();
    assertEquals(1, statements.size());
    assertThat(statements.get(0), equalToCompressingWhiteSpace(expectedProtoBundle));
    assertNotNull(ddl.hashCode());
  }

  @Test
  public void testDdlEquals() {
    Ddl ddl1 = Ddl.builder(Dialect.GOOGLE_STANDARD_SQL).build();
    Ddl ddl2 = Ddl.builder(Dialect.POSTGRESQL).build();
    assertFalse(ddl1.equals(ddl2));
    Ddl.Builder ddl1Builder =
        Ddl.builder().createTable("Users").column("id").int64().endColumn().endTable();
    ddl1Builder.createTable("Users");
    ddl1 = ddl1Builder.build();
    assertFalse(ddl1.equals(ddl2));
  }

  @Test
  public void testIndexColumnBuilder() {
    IndexColumnsBuilder indexColumnsBuilder = new IndexColumnsBuilder(null, null);
    assertThrows(IllegalArgumentException.class, () -> indexColumnsBuilder.name("name"));
    assertThrows(IllegalArgumentException.class, () -> indexColumnsBuilder.asc());
    assertThrows(IllegalArgumentException.class, () -> indexColumnsBuilder.desc());
    assertThrows(IllegalArgumentException.class, () -> indexColumnsBuilder.storing());
    assertThrows(IllegalArgumentException.class, () -> indexColumnsBuilder.nullsFirst());
    assertThrows(IllegalArgumentException.class, () -> indexColumnsBuilder.nullsLast());

    IndexColumn.Builder indexColumnBuilder = new AutoValue_IndexColumn.Builder();
    assertThrows(NullPointerException.class, () -> indexColumnBuilder.name(null));
    assertThrows(NullPointerException.class, () -> indexColumnBuilder.order(null));
    assertThrows(NullPointerException.class, () -> indexColumnBuilder.dialect(null));
    assertThrows(IllegalStateException.class, () -> indexColumnBuilder.autoBuild());
    IndexColumn indexColumn =
        indexColumnBuilder
            .name("col1")
            .order(Order.ASC)
            .dialect(Dialect.GOOGLE_STANDARD_SQL)
            .autoBuild();
    assertTrue(indexColumn.equals(indexColumn));
    assertFalse(indexColumn.equals(Boolean.TRUE));
    IndexColumn indexColumn1 = IndexColumn.create("col1", Order.ASC);
    assertTrue(indexColumn.equals(indexColumn1));
  }

  @Test
  public void testColumnBuilder() {
    Column.Builder columnBuilder = Column.builder();
    assertThrows(NullPointerException.class, () -> columnBuilder.name(null));
    assertThrows(NullPointerException.class, () -> columnBuilder.type(null));
    assertThrows(NullPointerException.class, () -> columnBuilder.columnOptions(null));
    assertThrows(NullPointerException.class, () -> columnBuilder.dialect(null));
    assertThrows(NullPointerException.class, () -> columnBuilder.name(null));
    assertThrows(NullPointerException.class, () -> columnBuilder.name(null));
    assertThrows(IllegalStateException.class, () -> columnBuilder.autoBuild());
    Column column = Column.builder().name("colName").type(Type.string()).autoBuild();
    assertTrue(column.equals(column));
    assertFalse(column.equals(Boolean.TRUE));
    Column column1 = Column.builder().name("colname").type(Type.bool()).autoBuild();
    assertFalse(column.equals(column1));
    column1 = Column.builder().name("colName").type(Type.bool()).autoBuild();
    assertFalse(column.equals(column1));
    column1 = Column.builder().name("colName").type(Type.string()).size(20).autoBuild();
    assertFalse(column.equals(column1));
    column1 = Column.builder().name("colName").type(Type.string()).notNull().autoBuild();
    assertFalse(column.equals(column1));
    column1 = Column.builder().name("colName").type(Type.string()).isGenerated(true).autoBuild();
    assertFalse(column.equals(column1));
    column1 =
        Column.builder()
            .name("colName")
            .type(Type.string())
            .generationExpression("1+2")
            .autoBuild();
    assertFalse(column.equals(column1));
    column1 =
        Column.builder().name("colName").type(Type.string()).defaultExpression("1+2").autoBuild();
    assertFalse(column.equals(column1));
    column1 = Column.builder().name("colName").type(Type.string()).isStored(true).autoBuild();
    assertFalse(column.equals(column1));
    column1 = Column.builder().name("colName").type(Type.string()).autoBuild();
    assertTrue(column.equals(column1));
  }

  @Test
  public void testTableBuilder() {
    Table.Builder tableBuilder = Table.builder();
    assertThrows(NullPointerException.class, () -> tableBuilder.primaryKeys(null));
    assertThrows(NullPointerException.class, () -> tableBuilder.columns(null));
    assertThrows(NullPointerException.class, () -> tableBuilder.indexes(null));
    assertThrows(NullPointerException.class, () -> tableBuilder.foreignKeys(null));
    assertThrows(NullPointerException.class, () -> tableBuilder.checkConstraints(null));
    assertThrows(NullPointerException.class, () -> tableBuilder.dialect(null));
    assertThrows(IllegalStateException.class, () -> tableBuilder.autoBuild());
    assertThrows(IllegalStateException.class, () -> tableBuilder.columns());
    Ddl.Builder ddlBuilder = Ddl.builder();
    ddlBuilder
        .createTable("Users")
        .column("id")
        .int64()
        .notNull()
        .endColumn()
        .primaryKey()
        .asc("id")
        .end()
        .endTable();
    Table table = ddlBuilder.build().table("Users");
    assertTrue(table.equals(table));
    assertFalse(table.equals(Boolean.TRUE));
    ddlBuilder = Ddl.builder();
    ddlBuilder
        .createTable("Users")
        .column("id")
        .int64()
        .notNull()
        .endColumn()
        .primaryKey()
        .asc("id")
        .end()
        .endTable();
    Table table1 = ddlBuilder.build().table("Users");
    assertTrue(table.equals(table1));
    tableBuilder.columns(ImmutableList.of());
    assertEquals(0, tableBuilder.columns().size());
  }

  @Test
  public void testForeignKeyBuilder() {
    ForeignKey.Builder foreignKeyBuilder = ForeignKey.builder();
    assertThrows(NullPointerException.class, () -> foreignKeyBuilder.name(null));
    assertThrows(NullPointerException.class, () -> foreignKeyBuilder.table(null));
    assertThrows(NullPointerException.class, () -> foreignKeyBuilder.referencedTable(null));
    assertThrows(NullPointerException.class, () -> foreignKeyBuilder.dialect(null));
    assertThrows(NullPointerException.class, () -> foreignKeyBuilder.referentialAction(null));
    // Setting null is OK for isEnforced since it's not supported for Postgres
    foreignKeyBuilder.isEnforced(null);
    assertThrows(IllegalStateException.class, () -> foreignKeyBuilder.build());
    ForeignKey foreignKey =
        foreignKeyBuilder.name("fk").table("table1").referencedTable("table2").build();
    assertTrue(foreignKey.equals(foreignKey));
    assertFalse(foreignKey.equals(Boolean.TRUE));
    ForeignKey foreignKey1 =
        ForeignKey.builder().name("fk").table("table1").referencedTable("table2").build();
    assertTrue(foreignKey.equals(foreignKey1));
  }

  @Test
  public void testForeignKeyBuilderActions() {
    ForeignKey fkBase =
        ForeignKey.builder().name("fk").table("Users").referencedTable("AllowedNames").build();
    ForeignKey.Builder fkWithDeleteCascade1Builder =
        ForeignKey.builder().name("fk_odc").table("Users").referencedTable("AllowedNames");
    fkWithDeleteCascade1Builder.columnsBuilder().add("first_name", "last_name");
    fkWithDeleteCascade1Builder.referencedColumnsBuilder().add("first_name", "last_name");
    fkWithDeleteCascade1Builder.referentialAction(Optional.of(ReferentialAction.ON_DELETE_CASCADE));
    ForeignKey fkWithDeleteCascade1 = fkWithDeleteCascade1Builder.build();
    assertTrue(fkWithDeleteCascade1.equals(fkWithDeleteCascade1));
    assertFalse(fkBase.equals(fkWithDeleteCascade1));

    ForeignKey.Builder fkWithDeleteCascade2Builder =
        ForeignKey.builder().name("fk_odc").table("Users").referencedTable("AllowedNames");
    fkWithDeleteCascade2Builder.columnsBuilder().add("first_name", "last_name");
    fkWithDeleteCascade2Builder.referencedColumnsBuilder().add("first_name", "last_name");
    fkWithDeleteCascade2Builder.referentialAction(Optional.of(ReferentialAction.ON_DELETE_CASCADE));
    ForeignKey fkWithDeleteCascade2 = fkWithDeleteCascade2Builder.build();
    assertTrue(fkWithDeleteCascade1.equals(fkWithDeleteCascade2));

    ForeignKey.Builder fkWithDeleteNoActionBuilder =
        ForeignKey.builder().name("fk_odna").table("Users").referencedTable("AllowedNames");
    fkWithDeleteNoActionBuilder.columnsBuilder().add("first_name", "last_name");
    fkWithDeleteNoActionBuilder.referencedColumnsBuilder().add("first_name", "last_name");
    fkWithDeleteNoActionBuilder.referentialAction(
        Optional.of(ReferentialAction.ON_DELETE_NO_ACTION));
    ForeignKey fkWithDeleteNoAction = fkWithDeleteNoActionBuilder.build();
    assertTrue(fkWithDeleteNoAction.equals(fkWithDeleteNoAction));
    assertFalse(fkWithDeleteCascade1.equals(fkWithDeleteNoAction));
  }

  @Test
  public void testUnsupportedForeignKeyBuilderActionThrowsError() {
    ForeignKey.Builder fkWithUnsupportedActionBuilder =
        ForeignKey.builder().name("fk_odsn").table("Users").referencedTable("AllowedNames");
    fkWithUnsupportedActionBuilder.columnsBuilder().add("first_name", "last_name");
    fkWithUnsupportedActionBuilder.referencedColumnsBuilder().add("first_name", "last_name");
    fkWithUnsupportedActionBuilder.referentialAction(
        Optional.of(ReferentialAction.ON_DELETE_SET_NULL));
    ForeignKey fkWithUnsupportedAction = fkWithUnsupportedActionBuilder.build();
    Throwable exception =
        assertThrows(IllegalArgumentException.class, () -> fkWithUnsupportedAction.prettyPrint());
    assertThat(exception.getMessage())
        .matches("Foreign Key action not supported: ON DELETE SET NULL");
  }

  @Test
  public void testCheckConstraintBuilder() {
    CheckConstraint.Builder checkConstraintBuilder = CheckConstraint.builder();
    assertThrows(NullPointerException.class, () -> checkConstraintBuilder.name(null));
    assertThrows(NullPointerException.class, () -> checkConstraintBuilder.expression(null));
    assertThrows(NullPointerException.class, () -> checkConstraintBuilder.dialect(null));
    assertThrows(IllegalStateException.class, () -> checkConstraintBuilder.build());
    CheckConstraint checkConstraint = checkConstraintBuilder.name("ck").expression("1<2").build();
    assertTrue(checkConstraint.equals(checkConstraint));
    assertFalse(checkConstraint.equals(Boolean.TRUE));
    CheckConstraint checkConstraint1 = checkConstraintBuilder.name("ck").expression("1<2").build();
    assertTrue(checkConstraint.equals(checkConstraint1));
  }

  @Test
  public void testNamedSchemaBuilder() {
    NamedSchema.Builder namedSchemaBuilder = NamedSchema.builder();
    namedSchemaBuilder.name("schema").dialect(Dialect.POSTGRESQL);
    NamedSchema namedSchema = namedSchemaBuilder.build();
    assertEquals("schema", namedSchema.name());
    assertEquals("CREATE SCHEMA \"schema\"", namedSchema.prettyPrint());

    NamedSchema.Builder namedSchemaBuilderG = NamedSchema.builder();
    namedSchemaBuilderG.name("schema").dialect(Dialect.GOOGLE_STANDARD_SQL);
    namedSchema = namedSchemaBuilderG.build();
    assertEquals("schema", namedSchema.name());
    assertEquals("CREATE SCHEMA `schema`", namedSchema.prettyPrint());
  }

  @Test
  public void testPropertyGraphOnView() {
    Ddl ddl =
        Ddl.builder()
            .createTable("GraphTableAccount")
            .column("loc_id")
            .int64()
            .endColumn()
            .column("aid")
            .int64()
            .endColumn()
            .column("owner_id")
            .int64()
            .endColumn()
            .column("name")
            .string()
            .max()
            .endColumn()
            .column("account_kind")
            .int64()
            .endColumn()
            .column("ProtoColumn")
            .bytes()
            .max()
            .endColumn()
            .column("generated_enum_field")
            .int64()
            .endColumn()
            .column("another_enum_field")
            .int64()
            .endColumn()
            .primaryKey()
            .asc("loc_id")
            .asc("aid")
            .end()
            .endTable()
            .createTable("GraphTablePerson")
            .column("loc_id")
            .int64()
            .endColumn()
            .column("pid")
            .int64()
            .endColumn()
            .primaryKey()
            .asc("loc_id")
            .asc("pid")
            .end()
            .endTable()
            .createView("V_FilteredPerson")
            .query("SELECT t.loc_id, t.pid FROM GraphTablePerson AS t WHERE t.loc_id = 1")
            .security(View.SqlSecurity.INVOKER)
            .endView()
            .createView("V_GroupByPerson")
            .query(
                "SELECT t.loc_id, t.pid, COUNT(*) AS cnt FROM GraphTablePerson AS t GROUP BY"
                    + " t.loc_id, t.pid ORDER BY cnt DESC")
            .security(View.SqlSecurity.INVOKER)
            .endView()
            .createPropertyGraph("aml_view_complex")
            .addNodeTable(
                GraphElementTable.builder()
                    .baseTableName("V_GroupByPerson")
                    .name("V_GroupByPerson")
                    .kind(GraphElementTable.Kind.NODE)
                    .keyColumns(ImmutableList.of("loc_id", "pid"))
                    .labelToPropertyDefinitions(
                        ImmutableList.of(
                            new LabelToPropertyDefinitions(
                                "V_GroupByPerson",
                                ImmutableList.of(
                                    new PropertyDefinition("loc_id", "loc_id"),
                                    new PropertyDefinition("pid", "pid"),
                                    new PropertyDefinition("cnt", "cnt")))))
                    .autoBuild())
            .addNodeTable(
                GraphElementTable.builder()
                    .baseTableName("V_FilteredPerson")
                    .name("V_FilteredPerson")
                    .kind(GraphElementTable.Kind.NODE)
                    .keyColumns(ImmutableList.of("loc_id", "pid"))
                    .labelToPropertyDefinitions(
                        ImmutableList.of(
                            new LabelToPropertyDefinitions(
                                "V_FilteredPerson",
                                ImmutableList.of(
                                    new PropertyDefinition("loc_id", "loc_id"),
                                    new PropertyDefinition("pid", "pid")))))
                    .autoBuild())
            .addNodeTable(
                GraphElementTable.builder()
                    .baseTableName("GraphTableAccount")
                    .name("GraphTableAccount")
                    .kind(GraphElementTable.Kind.NODE)
                    .keyColumns(ImmutableList.of("loc_id", "aid"))
                    .labelToPropertyDefinitions(
                        ImmutableList.of(
                            new LabelToPropertyDefinitions(
                                "GraphTableAccount",
                                ImmutableList.of(
                                    new PropertyDefinition("loc_id", "loc_id"),
                                    new PropertyDefinition("aid", "aid"),
                                    new PropertyDefinition("owner_id", "owner_id"),
                                    new PropertyDefinition("name", "name"),
                                    new PropertyDefinition("account_kind", "account_kind"),
                                    new PropertyDefinition("ProtoColumn", "ProtoColumn"),
                                    new PropertyDefinition(
                                        "generated_enum_field", "generated_enum_field"),
                                    new PropertyDefinition(
                                        "another_enum_field", "another_enum_field")))))
                    .autoBuild())
            .addEdgeTable(
                GraphElementTable.builder()
                    .baseTableName("GraphTableAccount")
                    .name("Owns")
                    .kind(GraphElementTable.Kind.EDGE)
                    .keyColumns(ImmutableList.of("loc_id", "aid"))
                    .sourceNodeTable(
                        new GraphNodeTableReference(
                            "V_FilteredPerson",
                            ImmutableList.of("loc_id", "pid"),
                            ImmutableList.of("loc_id", "owner_id")))
                    .targetNodeTable(
                        new GraphNodeTableReference(
                            "GraphTableAccount",
                            ImmutableList.of("loc_id", "aid"),
                            ImmutableList.of("loc_id", "aid")))
                    .labelToPropertyDefinitions(
                        ImmutableList.of(
                            new LabelToPropertyDefinitions(
                                "Owns",
                                ImmutableList.of(
                                    new PropertyDefinition("loc_id", "loc_id"),
                                    new PropertyDefinition("aid", "aid"),
                                    new PropertyDefinition("owner_id", "owner_id")))))
                    .autoBuild())
            .endPropertyGraph()
            .build();

    String expectedDdl =
        "CREATE TABLE `GraphTableAccount` (\n"
            + "\t`loc_id`                                INT64,\n"
            + "\t`aid`                                   INT64,\n"
            + "\t`owner_id`                              INT64,\n"
            + "\t`name`                                  STRING(MAX),\n"
            + "\t`account_kind`                          INT64,\n"
            + "\t`ProtoColumn`                           BYTES(MAX),\n"
            + "\t`generated_enum_field`                  INT64,\n"
            + "\t`another_enum_field`                    INT64,\n"
            + ") PRIMARY KEY (`loc_id` ASC, `aid` ASC)\n"
            + "CREATE TABLE `GraphTablePerson` (\n"
            + "\t`loc_id`                                INT64,\n"
            + "\t`pid`                                   INT64,\n"
            + ") PRIMARY KEY (`loc_id` ASC, `pid` ASC)\n"
            + "CREATE VIEW `V_FilteredPerson` SQL SECURITY INVOKER AS SELECT t.loc_id, t.pid FROM"
            + " GraphTablePerson AS t WHERE t.loc_id = 1\n"
            + "CREATE VIEW `V_GroupByPerson` SQL SECURITY INVOKER AS SELECT t.loc_id, t.pid,"
            + " COUNT(*) AS cnt FROM GraphTablePerson AS t GROUP BY t.loc_id, t.pid ORDER BY cnt"
            + " DESC\n"
            + "CREATE PROPERTY GRAPH aml_view_complex\n"
            + "NODE TABLES(\n"
            + "V_GroupByPerson AS V_GroupByPerson\n"
            + " KEY (loc_id, pid)\n"
            + "LABEL V_GroupByPerson PROPERTIES(loc_id, pid, cnt), V_FilteredPerson AS"
            + " V_FilteredPerson\n"
            + " KEY (loc_id, pid)\n"
            + "LABEL V_FilteredPerson PROPERTIES(loc_id, pid), GraphTableAccount AS"
            + " GraphTableAccount\n"
            + " KEY (loc_id, aid)\n"
            + "LABEL GraphTableAccount PROPERTIES(loc_id, aid, owner_id, name, account_kind,"
            + " ProtoColumn, generated_enum_field, another_enum_field))\n"
            + "EDGE TABLES(\n"
            + "GraphTableAccount AS Owns\n"
            + " KEY (loc_id, aid)\n"
            + "SOURCE KEY(loc_id, owner_id) REFERENCES V_FilteredPerson(loc_id,pid) DESTINATION"
            + " KEY(loc_id, aid) REFERENCES GraphTableAccount(loc_id,aid)\n"
            + "LABEL Owns PROPERTIES(loc_id, aid, owner_id))";
    assertThat(ddl.prettyPrint(), equalToCompressingWhiteSpace(expectedDdl));
  }

  @Test
  public void testPropertyGraphOnViewMixedOrder() {
    Ddl ddl =
        Ddl.builder()
            .createTable("Parts")
            .column("part_id")
            .int64()
            .endColumn()
            .column("part_name")
            .string()
            .max()
            .endColumn()
            .primaryKey()
            .asc("part_id")
            .end()
            .endTable()
            .createView("PartView")
            .query("SELECT part_id, part_name FROM Parts")
            .security(View.SqlSecurity.INVOKER)
            .endView()
            .createTable("Suppliers")
            .column("supplier_id")
            .int64()
            .endColumn()
            .column("supplier_name")
            .string()
            .max()
            .endColumn()
            .primaryKey()
            .asc("supplier_id")
            .end()
            .endTable()
            .createView("SupplierView")
            .query("SELECT supplier_id, supplier_name FROM Suppliers")
            .security(View.SqlSecurity.INVOKER)
            .endView()
            .createTable("PartSuppliers")
            .column("part_id")
            .int64()
            .endColumn()
            .column("supplier_id")
            .int64()
            .endColumn()
            .primaryKey()
            .asc("part_id")
            .asc("supplier_id")
            .end()
            .endTable()
            .createView("PartSuppliersView")
            .query("SELECT part_id, supplier_id FROM PartSuppliers")
            .security(View.SqlSecurity.INVOKER)
            .endView()
            .createPropertyGraph("SupplyChainGraph")
            .addNodeTable(
                GraphElementTable.builder()
                    .baseTableName("PartView")
                    .name("PartView")
                    .keyColumns(ImmutableList.of("part_id"))
                    .autoBuild())
            .addNodeTable(
                GraphElementTable.builder()
                    .baseTableName("SupplierView")
                    .name("SupplierView")
                    .keyColumns(ImmutableList.of("supplier_id"))
                    .autoBuild())
            .addEdgeTable(
                GraphElementTable.builder()
                    .baseTableName("PartSuppliersView")
                    .name("PartSuppliersView")
                    .kind(GraphElementTable.Kind.EDGE)
                    .keyColumns(ImmutableList.of("part_id", "supplier_id"))
                    .sourceNodeTable(
                        new GraphNodeTableReference(
                            "PartView", ImmutableList.of("part_id"), ImmutableList.of("part_id")))
                    .targetNodeTable(
                        new GraphNodeTableReference(
                            "SupplierView",
                            ImmutableList.of("supplier_id"),
                            ImmutableList.of("supplier_id")))
                    .autoBuild())
            .endPropertyGraph()
            .build();

    String expectedDdl =
        "CREATE TABLE `Parts` (\n"
            + "\t`part_id`                               INT64,\n"
            + "\t`part_name`                             STRING(MAX),\n"
            + ") PRIMARY KEY (`part_id` ASC)\n\n\n"
            + "CREATE TABLE `PartSuppliers` (\n"
            + "\t`part_id`                               INT64,\n"
            + "\t`supplier_id`                           INT64,\n"
            + ") PRIMARY KEY (`part_id` ASC, `supplier_id` ASC)\n\n\n"
            + "CREATE TABLE `Suppliers` (\n"
            + "\t`supplier_id`                           INT64,\n"
            + "\t`supplier_name`                         STRING(MAX),\n"
            + ") PRIMARY KEY (`supplier_id` ASC)\n\n\n"
            + "CREATE VIEW `PartSuppliersView` SQL SECURITY INVOKER AS SELECT part_id, supplier_id"
            + " FROM PartSuppliers\n"
            + "CREATE VIEW `PartView` SQL SECURITY INVOKER AS SELECT part_id, part_name FROM"
            + " Parts\n"
            + "CREATE VIEW `SupplierView` SQL SECURITY INVOKER AS SELECT supplier_id, supplier_name"
            + " FROM Suppliers\n"
            + "CREATE PROPERTY GRAPH SupplyChainGraph\n"
            + "NODE TABLES(\n"
            + "PartView AS PartView\n"
            + " KEY (part_id)\n"
            + ", SupplierView AS SupplierView\n"
            + " KEY (supplier_id)\n"
            + ")\n"
            + "EDGE TABLES(\n"
            + "PartSuppliersView AS PartSuppliersView\n"
            + " KEY (part_id, supplier_id)\n"
            + "SOURCE KEY(part_id) REFERENCES PartView(part_id) DESTINATION"
            + " KEY(supplier_id) REFERENCES SupplierView(supplier_id)\n"
            + ")";
    assertThat(ddl.prettyPrint(), equalToCompressingWhiteSpace(expectedDdl));
  }

  @Test
  public void testPropertyGraphOnViewTablesFirst() {
    Ddl ddl =
        Ddl.builder()
            .createTable("Parts2")
            .column("part_id")
            .int64()
            .endColumn()
            .column("part_name")
            .string()
            .max()
            .endColumn()
            .primaryKey()
            .asc("part_id")
            .end()
            .endTable()
            .createTable("Suppliers2")
            .column("supplier_id")
            .int64()
            .endColumn()
            .column("supplier_name")
            .string()
            .max()
            .endColumn()
            .primaryKey()
            .asc("supplier_id")
            .end()
            .endTable()
            .createTable("PartSuppliers2")
            .column("part_id")
            .int64()
            .endColumn()
            .column("supplier_id")
            .int64()
            .endColumn()
            .primaryKey()
            .asc("part_id")
            .asc("supplier_id")
            .end()
            .endTable()
            .createView("PartView2")
            .query("SELECT part_id, part_name FROM Parts2")
            .security(View.SqlSecurity.INVOKER)
            .endView()
            .createView("SupplierView2")
            .query("SELECT supplier_id, supplier_name FROM Suppliers2")
            .security(View.SqlSecurity.INVOKER)
            .endView()
            .createView("PartSuppliersView2")
            .query("SELECT part_id, supplier_id FROM PartSuppliers2")
            .security(View.SqlSecurity.INVOKER)
            .endView()
            .createPropertyGraph("SupplyChainGraph2")
            .addNodeTable(
                GraphElementTable.builder()
                    .baseTableName("PartView2")
                    .name("PartView2")
                    .keyColumns(ImmutableList.of("part_id"))
                    .autoBuild())
            .addNodeTable(
                GraphElementTable.builder()
                    .baseTableName("SupplierView2")
                    .name("SupplierView2")
                    .keyColumns(ImmutableList.of("supplier_id"))
                    .autoBuild())
            .addEdgeTable(
                GraphElementTable.builder()
                    .baseTableName("PartSuppliersView2")
                    .name("PartSuppliersView2")
                    .kind(GraphElementTable.Kind.EDGE)
                    .keyColumns(ImmutableList.of("part_id", "supplier_id"))
                    .sourceNodeTable(
                        new GraphNodeTableReference(
                            "PartView2", ImmutableList.of("part_id"), ImmutableList.of("part_id")))
                    .targetNodeTable(
                        new GraphNodeTableReference(
                            "SupplierView2",
                            ImmutableList.of("supplier_id"),
                            ImmutableList.of("supplier_id")))
                    .autoBuild())
            .endPropertyGraph()
            .build();

    String expectedDdl =
        "CREATE TABLE `Parts2` (\n"
            + "\t`part_id`                               INT64,\n"
            + "\t`part_name`                             STRING(MAX),\n"
            + ") PRIMARY KEY (`part_id` ASC)\n\n\n"
            + "CREATE TABLE `PartSuppliers2` (\n"
            + "\t`part_id`                               INT64,\n"
            + "\t`supplier_id`                           INT64,\n"
            + ") PRIMARY KEY (`part_id` ASC, `supplier_id` ASC)\n\n\n"
            + "CREATE TABLE `Suppliers2` (\n"
            + "\t`supplier_id`                           INT64,\n"
            + "\t`supplier_name`                         STRING(MAX),\n"
            + ") PRIMARY KEY (`supplier_id` ASC)\n\n\n"
            + "CREATE VIEW `PartSuppliersView2` SQL SECURITY INVOKER AS SELECT part_id,"
            + " supplier_id FROM PartSuppliers2\n"
            + "CREATE VIEW `PartView2` SQL SECURITY INVOKER AS SELECT part_id, part_name FROM"
            + " Parts2\n"
            + "CREATE VIEW `SupplierView2` SQL SECURITY INVOKER AS SELECT supplier_id,"
            + " supplier_name FROM Suppliers2\n"
            + "CREATE PROPERTY GRAPH SupplyChainGraph2\n"
            + "NODE TABLES(\n"
            + "PartView2 AS PartView2\n"
            + " KEY (part_id)\n"
            + ", SupplierView2 AS SupplierView2\n"
            + " KEY (supplier_id)\n"
            + ")\n"
            + "EDGE TABLES(\n"
            + "PartSuppliersView2 AS PartSuppliersView2\n"
            + " KEY (part_id, supplier_id)\n"
            + "SOURCE KEY(part_id) REFERENCES PartView2(part_id) DESTINATION"
            + " KEY(supplier_id) REFERENCES SupplierView2(supplier_id)\n"
            + ")";
    assertThat(ddl.prettyPrint(), equalToCompressingWhiteSpace(expectedDdl));
  }

  @Test
  public void testGraphOnViewWithNamedSchema() {
    Ddl ddl =
        Ddl.builder()
            .createSchema("Sch1")
            .endNamedSchema()
            .createSchema("Sch2")
            .endNamedSchema()
            .createTable("Sch1.Account")
            .column("AccountID")
            .int64()
            .notNull()
            .endColumn()
            .column("Money")
            .float64()
            .endColumn()
            .column("AnotherMoney")
            .float64()
            .endColumn()
            .primaryKey()
            .asc("AccountID")
            .end()
            .endTable()
            .createView("V0")
            .query("SELECT Account.AccountID, Account.Money FROM Sch1.Account")
            .security(View.SqlSecurity.INVOKER)
            .endView()
            .createView("Sch1.V1")
            .query("SELECT Account.AccountID, Account.Money FROM Sch1.Account")
            .security(View.SqlSecurity.INVOKER)
            .endView()
            .createView("Sch2.V2")
            .query("SELECT Account.AccountID, Account.Money FROM Sch1.Account")
            .security(View.SqlSecurity.INVOKER)
            .endView()
            .createPropertyGraph("aml")
            .addNodeTable(
                GraphElementTable.builder()
                    .baseTableName("V0")
                    .name("V0")
                    .keyColumns(ImmutableList.of("AccountID"))
                    .labelToPropertyDefinitions(
                        ImmutableList.of(
                            new LabelToPropertyDefinitions(
                                "V0",
                                ImmutableList.of(
                                    new PropertyDefinition("AccountID", "AccountID"),
                                    new PropertyDefinition("Money", "Money")))))
                    .autoBuild())
            .addNodeTable(
                GraphElementTable.builder()
                    .baseTableName("Sch1.V1")
                    .name("Sch1.V1")
                    .keyColumns(ImmutableList.of("AccountID"))
                    .labelToPropertyDefinitions(
                        ImmutableList.of(
                            new LabelToPropertyDefinitions(
                                "Sch1.V1",
                                ImmutableList.of(
                                    new PropertyDefinition("AccountID", "AccountID"),
                                    new PropertyDefinition("Money", "Money")))))
                    .autoBuild())
            .addNodeTable(
                GraphElementTable.builder()
                    .baseTableName("Sch2.V2")
                    .name("Sch2.V2")
                    .keyColumns(ImmutableList.of("AccountID"))
                    .labelToPropertyDefinitions(
                        ImmutableList.of(
                            new LabelToPropertyDefinitions(
                                "Sch2.V2",
                                ImmutableList.of(
                                    new PropertyDefinition("AccountID", "AccountID"),
                                    new PropertyDefinition("Money", "Money")))))
                    .autoBuild())
            .endPropertyGraph()
            .build();

    String expectedDdl =
        "\nCREATE SCHEMA `Sch1`\n"
            + "CREATE SCHEMA `Sch2`CREATE TABLE `Sch1`.`Account` (\n"
            + "\t`AccountID`                             INT64 NOT NULL,\n"
            + "\t`Money`                                 FLOAT64,\n"
            + "\t`AnotherMoney`                          FLOAT64,\n"
            + ") PRIMARY KEY (`AccountID` ASC)\n\n\n"
            + "CREATE VIEW `Sch1`.`V1` SQL SECURITY INVOKER AS SELECT Account.AccountID, Account.Money FROM Sch1.Account\n"
            + "CREATE VIEW `Sch2`.`V2` SQL SECURITY INVOKER AS SELECT Account.AccountID, Account.Money FROM Sch1.Account\n"
            + "CREATE VIEW `V0` SQL SECURITY INVOKER AS SELECT Account.AccountID, Account.Money FROM Sch1.Account\n"
            + "CREATE PROPERTY GRAPH aml\n"
            + "NODE TABLES(\n"
            + "V0 AS V0\n"
            + " KEY (AccountID)\n"
            + "LABEL V0 PROPERTIES(AccountID, Money), Sch1.V1 AS Sch1.V1\n"
            + " KEY (AccountID)\n"
            + "LABEL Sch1.V1 PROPERTIES(AccountID, Money), Sch2.V2 AS Sch2.V2\n"
            + " KEY (AccountID)\n"
            + "LABEL Sch2.V2 PROPERTIES(AccountID, Money))";
    assertThat(ddl.prettyPrint(), equalToCompressingWhiteSpace(expectedDdl));
  }
}
